#include <iostream>
#include <cstdlib>
#include <fstream>
#include <sstream>
#include <ctime>
#include <vector>
#include <map>
#include <string>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/signal.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/time.h>

#define SERV_PORT 9877
#define MAXPROCESS 100

using namespace std;

//structure for statistics
struct readData{
	long long totalLength;//total lenght currently read
	long long length;//Length of the leatest statistics
	long msec;//microseconds
	long sec;//seconds
	double rate;//rate of the leatest statistics
	double averageRate;//average rate from the beginning
	double totalTime;//total time from the beginning
	ofstream * rateFile;//record the rate
	ofstream * averageRateFile;//record the average rate
};

//contains each thread's statistics information
map<pthread_t,struct readData> statisticsData;
static ofstream sum;//for the total

static int BUFFERSIZE = 1024;//default buffersize = 1KB
static int PROCESS = 1;//default process number = 1
static int THREAD = 1;//default thread number = 1
static bool QUIET = false;//enable to print info defaultly
static int INTERVALTIME = 10000;//10ms
static int INTERVALTIMESEC = 0;//0s
char ipaddr[INET_ADDRSTRLEN];//hold the IP address

struct sockaddr_in servaddr;//server socket address
struct timeval start;

void statistics(int signo);//for SIGALRM
int argumenthandler(int argc,char **argv);//deal with the arguments
void setTimer();//set the alarm 
void initialThreadData(struct readData & threadReadData, struct timeval currentTime);
void *communication(void *arg);

int main(int argc,char **argv){
	pthread_t tid;
	pid_t pid;
	vector<pthread_t> joinList[MAXPROCESS];
	vector<pid_t> waitList;

	//deal with the arguments
	if(argumenthandler(argc,argv))
		return -1;
	
	sum.open("./data/sum");
	if(!sum.is_open()){
			perror("cannot open file sum");
			exit(-1);
	}

 	//fill in the socket address structure
 	bzero(&servaddr,sizeof(struct sockaddr_in));
	servaddr.sin_family = AF_INET;
	servaddr.sin_port = htons(SERV_PORT);
	if(inet_pton(AF_INET,ipaddr,&servaddr.sin_addr)<=0){
		perror("inet_pton error");
		exit(-1);
	}

	gettimeofday(&start,NULL);//get the starting time

	//start the process of reading data
	for(int i = 0;i < PROCESS;++i){
		if((pid = fork())==0){
			//create the threads and update the join list
			for(int j = 0;j < THREAD;++j){
				pthread_create(&tid,NULL,communication,(void *)(j+i*THREAD));
				joinList[i].push_back(tid);
			}

			//main process  waits for the child threads
			for(vector<pthread_t>::iterator it = joinList[i].begin();
				it!=joinList[i].end();++it)
				pthread_join(*it,NULL);
			break;
		}
		waitList.push_back(pid);
	}

	//main process waits the child processes
	for(int k = 0;k < PROCESS;++k)
		waitpid(waitList[k],NULL,0);

	exit(0);
}

void initialThreadData(struct readData &threadReadData,struct timeval currentTime){
	threadReadData.length = 0;
	threadReadData.totalLength = 0;
	threadReadData.totalTime = 0;
	threadReadData.sec = currentTime.tv_sec;
	threadReadData.msec = currentTime.tv_usec;
	threadReadData.rate = 0;
	threadReadData.averageRate = 0;
	threadReadData.rateFile = new ofstream();
	threadReadData.averageRateFile = new ofstream();
}

void *communication(void *arg){
	int sockfd;
	struct readData threadReadData;
	char buff[BUFFERSIZE+1];
	pthread_t myTid;
	stringstream output;

	myTid = pthread_self();
	signal(SIGALRM,statistics);//catch the SIGALRM signal
	setTimer();//set the alarm interrupt
	initialThreadData(threadReadData,start);//initiate data
	statisticsData.insert(make_pair(myTid,threadReadData));

	//create output files for rate and average rate 
	output.str("");
	output<<"./data/"<<(int)arg;
	threadReadData.rateFile->open(output.str().c_str());
	if(!threadReadData.rateFile->is_open()){
		cout<<"cannot open rate file: "<<output.str()<<endl;
		exit(-1);
	}
	output<<"avg";
	threadReadData.averageRateFile->open(output.str().c_str());
	if(!threadReadData.averageRateFile->is_open()){
		cout<<"cannot open average rate file: "<<output.str()<<endl;
		exit(-1);
	}

	//create socket
	if((sockfd = socket(AF_INET,SOCK_STREAM,0))<0){
		perror("socket error");
		exit(-1);
	}

	//connect to server
	if(connect(sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr))<0){
		perror("connect error");
		exit(-1);
	}

	//send request to server
	snprintf(buff,BUFFERSIZE,"START");
	if(send(sockfd,buff,strlen(buff),0)<0){
		perror("send error");
		exit(-1);
	}

	//get data
	ssize_t n;
	while((n = recv(sockfd,buff,BUFFERSIZE,0))>0){
		//increment the length
		statisticsData[myTid].length += (n/10);
		statisticsData[myTid].totalLength += (n/10);
	}

	if(n<0){
		perror("recv error");
		exit(-1);
	}

	cout<<"Client "<<(int)arg<<" ends"<<endl;
	close(sockfd);
	threadReadData.rateFile->close();
	threadReadData.averageRateFile->close();
	delete threadReadData.rateFile;
	delete threadReadData.averageRateFile;
}

void printUsage(){
	cout<<"Usage: ./client [-a IPaddr] [-t threadNum] [-p processNum] [-b buffersize] [-h]"<<endl
		<<"\t-a server IP address"<<endl
		<<"\t-m time interval microseconds between every alarm"<<endl
		<<"\t-s time interval seconds between every alarm"<<endl
		<<"\t-t set the thread number"<<endl
		<<"\t-p set the process number"<<endl
		<<"\t-b set the buffersize for every read or write"<<endl
		<<"\t-q disable to print the statistics information"<<endl
		<<"\t-h help"<<endl;
}

int argumenthandler(int argc,char **argv){
	int ch;
	while((ch = getopt(argc,argv,"m:t:b:hs:p:qa:"))!=-1){
		switch(ch){
			case 'a'://IP address
				strncpy(ipaddr,optarg,strlen(optarg));
				cout<<"Server IP: "<<ipaddr<<endl;
				break;
			case 'm'://time interval
				INTERVALTIME = atoi(optarg)*1000;
				break;
			case 't'://thread number
				THREAD = atoi(optarg);
				break;
			case 'p'://process number
				PROCESS = atoi(optarg);
				break;
			case 's'://time interval second
				INTERVALTIMESEC = atoi(optarg);
				break;
			case 'b'://buffer size
				BUFFERSIZE = atoi(optarg);
				cout<<"Buffersize: "<<BUFFERSIZE<<endl;
				break;
			case 'q'://disable to print info
				QUIET = true;
				break;
			case 'h'://help
				printUsage();
				return -1;
				break;
			default:
				cout<<"Unsupported argument: "<<(char)ch<<endl;
				printUsage();
				return -1;
		}
	}
	return 0;
}

void setTimer(){
	struct itimerval itimer;

	itimer.it_interval.tv_sec = INTERVALTIMESEC;
	itimer.it_interval.tv_usec = INTERVALTIME;
	itimer.it_value.tv_sec = INTERVALTIMESEC;
	itimer.it_value.tv_usec = INTERVALTIME;
	setitimer(ITIMER_REAL,&itimer,NULL);
}

void statistics(int signo){
	long long whole = 0;
	if(signo == SIGALRM){
		if(!QUIET){
			cout<<"Statistics ==>"
				<<"Process: "<<getpid()<<" Thread: "<<pthread_self()
				<<" StatisticsSize: "<<statisticsData.size()<<endl;
		}
		map<pthread_t,struct readData>::iterator it = statisticsData.begin();

		struct timeval sysTime;
		while(it!=statisticsData.end()){
			//get current time
			gettimeofday(&sysTime,NULL);
			double time = ((sysTime.tv_sec-it->second.sec)+
				(sysTime.tv_usec-it->second.msec)/1000000.0);

			//calculate rate
			it->second.rate = ((double)it->second.length*10/1024/1024)/time;

			//record the current time
			it->second.sec = sysTime.tv_sec;
			it->second.msec = sysTime.tv_usec;

			//record total time
			it->second.totalTime += time;

			//calculate average rate
			it->second.averageRate = ((double)it->second.totalLength*10/1024/1024)/it->second.totalTime;
			
			
			whole += it->second.totalLength;

			//print the statistics informaion
			if(!QUIET){
				cout<<"TotalSize:\t"<<it->second.totalLength*10/1024<<" KB\t"
					<<"\tAverageRate:\t"<<it->second.averageRate<<" MB/s"<<endl
					<<"ReadSize:\t"<<it->second.length*10/1024<<" KB\t"
					<<"\tRealtimeRate:\t"<<it->second.rate<<" MB/s"<<endl;
			}

			it->second.length = 0;//reset the length
			time_t t;
			struct tm *current;
			current = localtime(&t);
			*(it->second.rateFile)//<<"CurrentTime: "<<asctime(current)<<"\t"
				                  <<"CurrentRate: "<<it->second.rate<<" MB/s"<<endl;
			*(it->second.averageRateFile)//<<"CurrentTime: "<<asctime(current)<<"\t"
				                         <<"AverageRate: "<<it->second.averageRate<<" MB/s"<<endl;
			++it;
		}
		double avgrate,t;
		struct timeval here;
		gettimeofday(&here,NULL);
		t =  ((here.tv_sec-start.tv_sec)+
	            (here.tv_usec-start.tv_usec)/1000000.0);
		avgrate = (whole*10)/1024/1024/t;

		sum<<"Average Rate: "<<avgrate<<" MB/s"<<endl;

		//cout<<"Finish statistics"<<endl;
	}
}
